package com.lengxf.kafka.producer;

import com.fasterxml.jackson.annotation.JsonFormat;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;

import java.util.Date;
import java.util.concurrent.CompletableFuture;

@Slf4j
@Component
@AllArgsConstructor
public class Producer {

    private final static String topic_name = "data_topic";
    private final KafkaTemplate<String, String> kafkaTemplate;

    @SneakyThrows
    public void sendMsg(String msg) {

        Msg msss = new Msg();
        msss.setA(new Date());
        msss.setB(1);
        msss.setC("c");

        ObjectMapper objectMapper = new ObjectMapper();
        String json = objectMapper.writeValueAsString(msss);

        CompletableFuture<SendResult<String, String>> send = kafkaTemplate.send(topic_name, json);
        SendResult<String, String> stringStringSendResult = send.get();
        RecordMetadata recordMetadata = stringStringSendResult.getRecordMetadata();
//        log.info("recordMetadata : {}", recordMetadata);

    }

    @Data
    public static class Msg {
        @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss", timezone = "GMT+8")
        private Date a;
        private Integer b;
        private String c;
    }

}
